package org.eventime

import org.FlinkStreamApp
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * description ：抽取时间戳和插入水位线
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/23 11:55
 * modified By ：
 * version:    : 1.0
 */
object EventTimeExample extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    val stream = environment.socketTextStream("hadoop101", 9999, '\n')
    stream.map(
      l => {
        val splits = l.split("\\s")
        (splits(0), splits(1).toLong * 1000)
      }
    )
      .assignTimestampsAndWatermarks(
        // 设置最大延迟时间5s
        new BoundedOutOfOrdernessTimestampExtractor[(String, Long)](Time.seconds(5)) {
          override def extractTimestamp(element: (String, Long)) = element._2
        }
      )
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      .process(new WatermarkProcessFunction())
      .print()
  }

  // [IN, OUT, KEY, W <: Window]
  class WatermarkProcessFunction extends ProcessWindowFunction[(String, Long), String, String, TimeWindow] {
    override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[String]) = {
      out.collect("窗口中共有： " + elements.size.toString + " 条数据")
    }
  }

}
